package com.hexinfo.dmpro.data.disaster.recovery.backups.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hexinfo.dmpro.common.model.ClusterSparingConf;
import com.hexinfo.dmpro.common.service.ClusterSparingConfService;
import com.hexinfo.dmpro.common.utils.CommonConstants;
import com.hexinfo.dmpro.component.common.service.impl.HiveQueryService;
import com.hexinfo.dmpro.component.common.util.DataSourceUtil;
import com.hexinfo.dmpro.component.data.transfer.model.DataExport;
import com.hexinfo.dmpro.data.disaster.recovery.backups.handler.IDataDisasterRecoveryHandler;
import com.hexinfo.dmpro.data.disaster.recovery.backups.model.DisasterRecoveryConf;
import com.hexinfo.dmpro.data.disaster.recovery.backups.dao.DisasterRecoveryConfMapper;
import com.hexinfo.dmpro.data.disaster.recovery.backups.model.DisasterRecoveryParam;
import com.hexinfo.dmpro.data.disaster.recovery.backups.service.DisasterRecoveryConfService;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;

/**
 * 生产数据灾备配置表
 *
 * @author wz.li
 * @date 2023-10-09 18:48:56
 */
@Slf4j
@Service
public class DisasterRecoveryConfServiceImpl extends ServiceImpl<DisasterRecoveryConfMapper, DisasterRecoveryConf> implements DisasterRecoveryConfService {

    @Autowired
    private IDataDisasterRecoveryHandler dataDisasterRecoveryHandler;

    @Autowired
    private ClusterSparingConfService clusterSparingConfService;

    @Autowired
    private HiveQueryService hiveQueryService;

    @Override
    public Page queryList(Page page, DisasterRecoveryConf disasterRecoveryConf) {
        LambdaQueryWrapper<DisasterRecoveryConf> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(DisasterRecoveryConf::getDel, CommonConstants.STATUS_NORMAL);
        Page returnPage = this.baseMapper.selectPage(page, wrapper);
        return returnPage;
    }

    /*
     * @Method: runById <br>
     * @Param: [id] <br>
     * @Return: void <br>
     * @Description：手动执行冷备流程<br>
     * @Author： wz.li<br>
     * @Date： 2023/10/10 16:30 <br>
     * @Version： V2.0.2<br>
     */
    @Override
    public void runById(String id) {
        try {
            DisasterRecoveryConf disasterRecoveryConf = this.getById(id);
            ClusterSparingConf clusterSparingConf = getClusterSparingConf();
            // TODO 创建冷备任务信息
            for (String dbName : getDatabaseList()) {
                for (String tableName : getTableList(dbName)) {
                    // TODO 创建冷备任务明细-表数据信息
                    disasterRecoveryByTable(disasterRecoveryConf, clusterSparingConf, dbName, tableName);
                }
            }
        } catch (Exception e) {

        }
    }

    /*
     * @Method: getClusterSparingConf <br>
     * @Param: [] <br>
     * @Return: com.hexinfo.dmpro.common.model.ClusterSparingConf <br>
     * @Description：获取万国集群配置信息<br>
     * @Author： wz.li<br>
     * @Date： 2023/10/11 16:46 <br>
     * @Version： V2.0.2<br>
     */
    private ClusterSparingConf getClusterSparingConf() {
        ClusterSparingConf clusterSparingConf = new ClusterSparingConf();
        clusterSparingConf.setClusterName("万国");
        return clusterSparingConfService.queryOne(clusterSparingConf);
    }

    /*
     * @Method: getDatabaseList <br>
     * @Param: [] <br>
     * @Return: java.util.List<java.lang.String> <br>
     * @Description：获取万国集群所有HIVE库集合<br>
     * @Author： wz.li<br>
     * @Date： 2023/10/11 16:43 <br>
     * @Version： V2.0.2<br>
     */
    private List<String> getDatabaseList() {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getHiveTargetConnection();
            return hiveQueryService.getDbList(connection);
        } catch (Exception e) {
            log.error("[生成数据冷备]-获取数据库列表失败，请检查配置，异常信息: {}", e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("获取数据库列表失败，请检查配置，异常信息: {}", e.getMessage()));
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    /*
     * @Method: getTableList <br>
     * @Param: [dbName] <br>
     * @Return: java.util.List<java.lang.String> <br>
     * @Description：获取指定库的表集合<br>
     * @Author： wz.li<br>
     * @Date： 2023/10/11 16:46 <br>
     * @Version： V2.0.2<br>
     */
    private List<String> getTableList(String dbName) {
        Connection connection = null;
        try {
            connection = DataSourceUtil.getHiveTargetConnection();
            return hiveQueryService.getTableList(connection, dbName);
        } catch (Exception e) {
            log.error("[生成数据冷备]-获取数据库[{}]表列表失败，请检查配置，异常信息: {}", dbName, e.getMessage(), e);
            throw new RuntimeException(StrUtil.format("获取数据库[{}]表列表失败，请检查配置，异常信息: {}", dbName, e.getMessage()));
        } finally {
            DataSourceUtil.close(connection);
        }
    }

    private void disasterRecoveryByTable(DisasterRecoveryConf disasterRecoveryConf,
                                         ClusterSparingConf clusterSparingConf, String database, String table) {
        log.info("查询{}.{}表的分区值", database, table);
        List<String> bizDateList = getTablePartition(clusterSparingConf, database, table);
        if (CollectionUtil.isNotEmpty(bizDateList) && 1 == 1) {
            for (String bizDate : bizDateList) {
                String hiveQuerySql = "select * from {}.{} where {} = '{}'";
                dataDisasterRecoveryHandler.disasterRecovery(new DisasterRecoveryParam());
            }
        } else {
            String hiveQuerySql = StrUtil.format("select * from {}.{}", database, table);
            dataDisasterRecoveryHandler.disasterRecovery(buildDisasterRecoveryParam(disasterRecoveryConf, hiveQuerySql, clusterSparingConf, database, table));
        }
    }

    private List<String> getTablePartition(ClusterSparingConf clusterSparingConf, String dbName, String tableName) {
        try {

            Configuration conf = new Configuration();
            conf.set("fs.defaultFS", "hdfs://172.21.33.125:8020");
            conf.set("hadoop.job.ugi", "hive");
            FileSystem fs = FileSystem.get(conf);
            Path directoryPath = new Path("/bigdata/cdh6/user/hive/warehouse/ods.db/test_1");
            List<String> fileList = getAllFiles(fs, directoryPath);
            System.out.println("======================>");
            for (String s : fileList) {
//                System.out.println(s);
            }
            // 关闭文件系统对象
            fs.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new ArrayList<>();
    }

    private static List<String> getAllFiles(FileSystem fs, Path path) throws Exception {
        // 获取目录下的文件列表
        FileStatus[] fileStatuses = fs.listStatus(path);
        List<String> fileList = new ArrayList<>();
        // 遍历文件列表
        for (FileStatus fileStatus : fileStatuses) {
            // 如果是子目录，则递归调用getAllFiles方法
            if (fileStatus.isDirectory()) {
                fileList.addAll(getAllFiles(fs, fileStatus.getPath()));
            } else {
                // 输出文件路径
                System.out.println(fileStatus.getPath() + "  " + DateUtil.formatDateTime(DateUtil.date(fileStatus.getModificationTime())));
                fileList.add(fileStatus.getPath().toString());
            }
        }
        return fileList;
    }

    private DisasterRecoveryParam buildDisasterRecoveryParam(DisasterRecoveryConf disasterRecoveryConf, String querySql,
                                            ClusterSparingConf clusterSparingConf, String database, String table){
        DisasterRecoveryParam disasterRecoveryParam = new DisasterRecoveryParam();
        disasterRecoveryParam.setHdfsUrls(clusterSparingConf.getHdfsPrefixPath());
        disasterRecoveryParam.setLocalDir(disasterRecoveryConf.getLocalDir());
        disasterRecoveryParam.setHiveQuerySql(querySql);
        disasterRecoveryParam.setDatabaseName(database);
        disasterRecoveryParam.setTableName(table);
        disasterRecoveryParam.setSeparator(disasterRecoveryConf.getSeparator());
        return disasterRecoveryParam;
    }
}
